package com.bw.app.dws;

import com.bw.app.func.KeywordUDTF;
import com.bw.bean.KeywordStats;
import com.bw.common.GmallConstant;
import com.bw.utils.ClickHouseUtil;
import com.bw.utils.MyKafkaUtil;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @ProjectName: BigData
 * @Package: com.bw.app.dws
 * @ClassName: KeywordStatsApp
 * @Author: Gy
 * @Description:
 * @Date: 2021/11/16 11:26
 */
public class KeywordStatsApp {
    public static void main(String[] args) throws Exception {
        //todo 1.基本环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

        EnvironmentSettings settings = EnvironmentSettings.newInstance().inStreamingMode().build();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);

        //todo 2.注册自定义函数
        tableEnv.createTemporarySystemFunction("ik_analyze", KeywordUDTF.class);

        //todo 3.创建动态表
        //3.1 声明主题以及消费者组
        String pageViewSourceTopic = "dwd_page_log";
        String groupId = "keywordstats_app_group";
        //3.2 建表
        tableEnv.executeSql(
                "CREATE TABLE page_view (" +
                        " common MAP<STRING, STRING>," +
                        " page MAP<STRING, STRING>," +
                        " ts BIGINT," +
                        " rowtime as TO_TIMESTAMP(FROM_UNIXTIME(ts/1000,'yyyy-MM-dd HH:mm:ss'))," +
                        " WATERMARK FOR rowtime AS rowtime - INTERVAL '2' SECOND) " +
                        " WITH (" + MyKafkaUtil.getKafkaDDl(pageViewSourceTopic, groupId) + ")"
        );

        //todo 4.从动态表中查询数据
        Table fullwordTable = tableEnv.sqlQuery(
                "select page['item'] fullword,rowtime " +
                        "from page_view " +
                        "where page['page_id']='good_list' and page['item'] IS NOT NULL");

        //todo 5.利用自定义函数   对搜索关键词进行拆分
        Table keywordTable = tableEnv.sqlQuery(
                "SELECT keyword,rowtime " +
                        "FROM " + fullwordTable + "," +
                        "LATERAL TABLE(ik_analyze(fullword)) AS t(keyword)");

        //todo 6.分组、开窗、聚合
        Table reduceTable = tableEnv.sqlQuery(
                "select keyword,count(*) ct,'" + GmallConstant.KEYWORD_SEARCH + "' source," +
                        "DATE_FORMAT(TUMBLE_START(rowtime,INTERVAL '10' SECOND),'yyyy-MM-dd HH:mm:ss') stt," +
                        "DATE_FORMAT(TUMBLE_END(rowtime,INTERVAL '10' SECOND),'yyyy-MM-dd HH:mm:ss') edt," +
                        "UNIX_TIMESTAMP()*1000 ts from " + keywordTable +
                        " group by TUMBLE(rowtime,INTERVAL '10' SECOND),keyword");

        //todo 7.转换为流
        DataStream<KeywordStats> keywordStatsDS = tableEnv.toAppendStream(reduceTable, KeywordStats.class);
        keywordStatsDS.print(">>>>>");

        //todo 8.写入ClickHouse
        keywordStatsDS.addSink(
                ClickHouseUtil.getJdbcSink("insert into  keyword_stats_2021(keyword,ct,source,stt,edt,ts) values(?,?,?,?,?,?)")
        );

        env.execute();
    }
}

